#pragma once

#include "my_disk_layout.h"
#include "my_disk_vinfo.h"
#include "comm/my_defs.h"
#include "comm/my_spin_lock.h"
#include "comm/my_mem.h"
#include "comm/my_buf.h"
#include "comm/my_rbtree.h"
#include "comm/my_queue.h"
#include "cache/util/my_cache_err.h"

typedef struct my_disk_buf_s my_disk_buf_t;

struct my_disk_buf_s {
    my_spin_lock_t              lock;

    my_queue_t                  queue;      // list in buf_pool or pending flush list.
    my_rbtree_node_t            tree_node;  // for rbtree            

    char                        *buf;
    uint8_t                     write_serial;
    int64_t                     start_off;  // offset inside the volume
    int64_t                     write_off;  // offset inside the buffer

    my_atomic_int_t             read_counter;
    my_atomic_int_t             write_counter;
};

static inline void
my_disk_buf_alloc_source(my_disk_buf_t *buf) {
    bzero(buf, sizeof(my_disk_buf_t));
    buf->buf = my_malloc(MY_DISK_BLOCK_SIZE);
    my_atomic_int_init(&buf->read_counter);
    my_atomic_int_init(&buf->write_counter);
}

static inline void
my_disk_buf_free_source(my_disk_buf_t *buf) {
    if (buf->buf) {
        my_free(buf->buf);
    }
}

static inline my_bool_t
my_disk_buf_can_clean(my_disk_buf_t *buf) {
    return my_atomic_int_load(&buf->read_counter) == 0;
}

static inline my_bool_t
my_disk_buf_can_flush(my_disk_buf_t *buf) {
    return my_atomic_int_load(&buf->write_counter) == 0;
}

static inline int64_t
my_disk_buf_data_len(my_disk_buf_t *buf) {
    int64_t res;
    my_spin_lock_lock(&buf->lock);
    res = buf->write_off;
    my_spin_lock_unlock(&buf->lock);
    return res;
}

void
my_disk_buf_clean(my_disk_buf_t *buf);

my_cache_err_t    // for async write
my_disk_buf_write_hold(my_disk_buf_t *buf, int64_t aligned_size, 
    int64_t *offset_in_buffer);

void // async write
my_disk_buf_write(my_disk_buf_t *buf, my_cache_key_t *key,
    my_buf_chain_t *data, uint32_t crc32, int64_t offset_in_buffer);

my_cache_err_t // for async read
my_disk_buf_read_hold(my_disk_buf_t *buf, uint8_t write_serial,
    int64_t offset, int64_t size);

my_cache_err_t   // async read
my_disk_buf_read(my_disk_buf_t *buf, my_cache_key_t *key, int64_t offset,
    int64_t size, my_buf_chain_t *data);


typedef struct my_disk_buf_pool_s my_disk_buf_pool_t;

struct my_disk_buf_pool_s {
    my_spin_lock_t          lock;
    my_queue_t              head;
    my_disk_buf_t           *bufs;
    int64_t                 total_bufs;
    int64_t                 cur_bufs;
};

static inline void
my_disk_buf_pool_init(my_disk_buf_pool_t *pool) {
    bzero(pool, sizeof(my_disk_buf_pool_t));
}

static inline void
my_disk_buf_pool_open(my_disk_buf_pool_t *pool, int64_t nbufs) {
    my_spin_lock_init(&pool->lock);
    my_queue_init(&pool->head);

    my_disk_buf_t *bufs = my_malloc(sizeof(my_disk_buf_t) * nbufs);
    for (int i = 0; i < nbufs; i++) {
        my_disk_buf_alloc_source(bufs + i);
        my_queue_insert_head(&pool->head, &(bufs+i)->queue);
    }

    pool->total_bufs = nbufs;
    pool->cur_bufs = nbufs;
}

static inline void
my_disk_buf_pool_close(my_disk_buf_pool_t *pool) {
    if (!pool->bufs) {
        return;
    }

    for (int i = 0; i < pool->total_bufs; i++) {
        my_disk_buf_free_source(pool->bufs + i);
    }
    my_free(pool->bufs);
}

my_disk_buf_t *
my_disk_buf_pool_alloc(my_disk_buf_pool_t *pool, uint8_t write_serial,
    int64_t start_offset);

void
my_disk_buf_pool_free(my_disk_buf_pool_t *pool, my_disk_buf_t *buf);